package com.jdjc.subject.infra.basic.es;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.http.HttpHost;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.*;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.sort.ScoreSortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.stereotype.Component;


import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;

/**
 * 自定义的es工具类
 */
@Component
@Slf4j
public class EsRestClient {
    //RestHighLevelClient 是 Elasticsearch 提供的一个 Java API，用于通过 Java 方法和数据结构与 Elasticsearch 集群和索引进行交互，而不是直接使用 HTTP 请求。
    // 定义一个静态的Map，用于存储多个RestHighLevelClient实例，每个实例通过一个唯一的字符串名称进行索引
    private static Map<String, RestHighLevelClient> clientMap = new HashMap<>();

    // 使用@Resource注解注入EsConfigProperties类型的bean，该类包含了es集群名称和集群节点
    @Resource
    private EsConfigProperties esConfigProperties;

    private static final RequestOptions COMMON_OPTIONS;


    static {
        // 创建一个RequestOptions构建器，从默认的RequestOptions开始
        // 这允许你修改默认设置，如超时时间、重试策略等
        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
        // 使用构建器构建一个新的RequestOptions实例，并将这个实例赋值给COMMON_OPTIONS常量
        // 这个实例将包含你通过构建器设置的所有自定义配置
        COMMON_OPTIONS = builder.build();
    }

    // 使用@PostConstruct注解标记的方法会在Bean初始化时执行，用于执行一些初始化操作
    @PostConstruct
    public void initialize() {
        // 获取配置文件中定义的Elasticsearch集群配置列表
        List<EsClusterConfig> esConfigs = esConfigProperties.getEsConfigs();
        // 遍历每个Elasticsearch集群配置
        for (EsClusterConfig esConfig : esConfigs) {
            // 输出日志信息，显示当前正在初始化的集群名称和节点信息
            log.info("Initialize:config:name:{},node:{} ", esConfig.getName(), esConfig.getNodes());
            // 根据配置信息初始化RestHighLevelClient实例
            RestHighLevelClient restHighLevelClient = initRestClient(esConfig);
            // 如果初始化成功（即restHighLevelClient不为null），则将其存储在clientMap中
            if (restHighLevelClient != null) {
                clientMap.put(esConfig.getName(), restHighLevelClient);
            } else {
                // 如果初始化失败，输出错误日志
                log.error("config:name:{},node:{}.initError", esConfig.getName(), esConfig.getNodes());
            }
        }
    }

    // 根据传入的EsClusterConfig配置信息初始化RestHighLevelClient实例
    private RestHighLevelClient initRestClient(EsClusterConfig esConfig) {
        // 将配置中的节点信息（以逗号分隔）分割成字符串数组
        String[] ipPortArr = esConfig.getNodes().split(",");
        // 创建一个HttpHost列表，用于存储从ipPortArr解析出的节点信息
        List<HttpHost> httpHostList = new ArrayList<>(ipPortArr.length);
        // 遍历每个节点信息
        for (String ipPort : ipPortArr) {
            // 将单个节点信息（如"192.168.1.1:9200"）分割成IP和端口
            String[] ipPortInfo = ipPort.split(":");
            if (ipPortInfo.length == 2) {
                // 创建HttpHost实例，并将其添加到httpHostList中
                HttpHost httpHost = new HttpHost(ipPortInfo[0], NumberUtils.toInt(ipPortInfo[1]));
                httpHostList.add(httpHost);
            }
        }
        // 将httpHostList转换成数组
        HttpHost[] httpHosts = new HttpHost[httpHostList.size()];
        httpHostList.toArray(httpHosts);

        // 创建RestClientBuilder实例，传入节点信息数组
        RestClientBuilder builder = RestClient.builder(httpHosts);
        // 创建并返回RestHighLevelClient实例
        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(builder);
        return restHighLevelClient;
    }

    /**
     * 获取与指定集群名称对应的RestHighLevelClient实例
     *
     * @param clusterName
     * @return
     */
    private static RestHighLevelClient getClient(String clusterName) {
        return clientMap.get(clusterName);
    }

    /**
     * 新增文档
     */
    public static boolean insertDoc(EsIndexInfo esIndexInfo, EsSourceData esSourceData) {
        try {
            // 创建一个IndexRequest对象，指定要操作的索引名称
            IndexRequest indexRequest = new IndexRequest(esIndexInfo.getIndexName());
            // 设置要索引的文档的内容，通常是一个Map或JSON字符串
            indexRequest.source(esSourceData.getData());
            // 设置文档的ID，如果没有指定，Elasticsearch会自动生成一个
            indexRequest.id(esSourceData.getDocId());
            // 获取与指定集群名称对应的RestHighLevelClient实例，然后执行索引操作，将文档添加到Elasticsearch索引中
            getClient(esIndexInfo.getClusterName()).index(indexRequest, COMMON_OPTIONS);
            return true;
        } catch (Exception e) {
            log.error("insertDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 更新文档
     *
     * @param esIndexInfo
     * @param esSourceData
     * @return
     */
    public static boolean updateDoc(EsIndexInfo esIndexInfo, EsSourceData esSourceData) {
        try {
            UpdateRequest updateRequest = new UpdateRequest();
            updateRequest.index(esIndexInfo.getIndexName());
            updateRequest.id(esSourceData.getDocId());
            updateRequest.doc(esSourceData.getData());
            getClient(esIndexInfo.getClusterName()).update(updateRequest, COMMON_OPTIONS);
            return true;
        } catch (Exception e) {
            log.error("updateDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 批量更新文档
     *
     * @param esIndexInfo
     * @param esSourceDataList
     * @return
     */
    public static boolean batchUpdateDoc(EsIndexInfo esIndexInfo,
                                         List<EsSourceData> esSourceDataList) {
        try {
            // 初始化一个布尔变量flag，用于标识是否有更新请求被添加到批量请求中
            boolean flag = false;
            // 创建一个BulkRequest对象，用于构建批量请求
            BulkRequest bulkRequest = new BulkRequest();
            for (EsSourceData esSourceData : esSourceDataList) {
                // 获取当前EsSourceData对象的文档ID
                String docId = esSourceData.getDocId();
                // 如果文档ID非空，则进行更新操作
                if (StringUtils.isNotBlank(docId)) {
                    // 创建一个UpdateRequest对象，用于构建更新请求
                    UpdateRequest updateRequest = new UpdateRequest();
                    // 设置更新请求的索引名称
                    updateRequest.index(esIndexInfo.getIndexName());
                    // 设置更新请求的文档ID
                    updateRequest.id(esSourceData.getDocId());
                    // 设置更新请求的文档内容
                    updateRequest.doc(esSourceData.getData());
                    // 将更新请求添加到批量请求中
                    bulkRequest.add(updateRequest);
                    // 设置flag为true，表示有更新请求被添加到批量请求中
                    flag = true;
                }
            }
            // 如果flag为true，即有更新请求被添加到批量请求中，则执行批量更新操作
            if (flag) {
                BulkResponse bulk = getClient(esIndexInfo.getClusterName()).bulk(bulkRequest, COMMON_OPTIONS);
                // 如果批量更新操作中有失败的请求，则返回false
                if (bulk.hasFailures()) {
                    return false;
                }
            }
            return true;
        } catch (Exception e) {
            log.error("batchUpdateDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }


    /**
     * 删除所有文档
     *
     * @param esIndexInfo
     * @return
     */
    public static boolean delete(EsIndexInfo esIndexInfo) {
        try {
            // 创建一个DeleteByQueryRequest对象，指定要操作的索引名称
            DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(esIndexInfo.getIndexName());
            // 使用matchAllQuery()方法来匹配索引中的所有文档
            deleteByQueryRequest.setQuery(QueryBuilders.matchAllQuery());
            // 执行删除操作，获取与指定集群名称对应的RestHighLevelClient实例，然后执行删除请求，返回一个BulkByScrollResponse对象，包含了删除操作的结果
            BulkByScrollResponse response = getClient(esIndexInfo.getClusterName()).deleteByQuery(deleteByQueryRequest, COMMON_OPTIONS);
            // 获取删除的文档数量
            long deleted = response.getDeleted();
            log.info("deleted.size:{}", deleted);
            return true;
        } catch (Exception e) {
            log.error("delete.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 根据docId删除指定文档
     *
     * @param esIndexInfo
     * @param docId
     * @return
     */
    public static boolean deleteDoc(EsIndexInfo esIndexInfo, String docId) {
        try {
            DeleteRequest deleteRequest = new DeleteRequest(esIndexInfo.getIndexName());
            deleteRequest.id(docId);
            DeleteResponse response = getClient(esIndexInfo.getClusterName()).delete(deleteRequest, COMMON_OPTIONS);
            log.info("deleteDoc.response:{}", JSON.toJSONString(response));
            return true;
        } catch (Exception e) {
            log.error("deleteDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 根据docId判断文档是否存在
     *
     * @param esIndexInfo
     * @param docId
     * @return
     */
    public static boolean isExistDocById(EsIndexInfo esIndexInfo, String docId) {
        try {
            GetRequest getRequest = new GetRequest(esIndexInfo.getIndexName());
            getRequest.id(docId);
            return getClient(esIndexInfo.getClusterName()).exists(getRequest, COMMON_OPTIONS);
        } catch (Exception e) {
            log.error("isExistDocById.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 根据docID获取文档
     *
     * @param esIndexInfo
     * @param docId
     * @return
     */
    public static Map<String, Object> getDocById(EsIndexInfo esIndexInfo, String docId) {
        try {
            GetRequest getRequest = new GetRequest(esIndexInfo.getIndexName());
            getRequest.id(docId);
            GetResponse response = getClient(esIndexInfo.getClusterName()).get(getRequest, COMMON_OPTIONS);
            Map<String, Object> source = response.getSource();
            return source;
        } catch (Exception e) {
            log.error("isExistDocById.exception:{}", e.getMessage(), e);
        }
        return null;
    }

    /**
     * 根据特定字段匹配文档
     *
     * @param esIndexInfo
     * @param docId
     * @param fields
     * @return
     */
    public static Map<String, Object> getDocById(EsIndexInfo esIndexInfo, String docId,
                                                 String[] fields) {
        try {
            GetRequest getRequest = new GetRequest(esIndexInfo.getIndexName());
            getRequest.id(docId);
            //指定在执行搜索操作时是否以及如何获取文档的源字段。第一个参数启用源字段的获取。第二个参数 指定要获取的源字段的名称，如果为 null，则获取所有字段。第三个参数用于指定要排除的字段
            FetchSourceContext fetchSourceContext = new FetchSourceContext(true, fields, null);
            getRequest.fetchSourceContext(fetchSourceContext);
            GetResponse response = getClient(esIndexInfo.getClusterName()).get(getRequest, COMMON_OPTIONS);
            Map<String, Object> source = response.getSource();
            return source;
        } catch (Exception e) {
            log.error("isExistDocById.exception:{}", e.getMessage(), e);
        }
        return null;
    }

    /**
     * 搜索
     *
     * @param esIndexInfo
     * @param esSearchRequest
     * @return
     */
    public static SearchResponse searchWithTermQuery(EsIndexInfo esIndexInfo,
                                                     EsSearchRequest esSearchRequest) {
        try {
            //查询条件
            BoolQueryBuilder bq = esSearchRequest.getBq();
            //基于哪些字段进行搜索
            String[] fields = esSearchRequest.getFields();
            //分页页数
            int from = esSearchRequest.getFrom();
            //一次抓取条数
            int size = esSearchRequest.getSize();
            //快照缓存时间
            Long minutes = esSearchRequest.getMinutes();
            //是否需要快照
            Boolean needScroll = esSearchRequest.getNeedScroll();
            //排序字段
            String sortName = esSearchRequest.getSortName();
            //排序类型
            SortOrder sortOrder = esSearchRequest.getSortOrder();

            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.query(bq);
            searchSourceBuilder.fetchSource(fields, null).from(from).size(size);

            //在搜索结果中启用和配置高亮显示
            if (Objects.nonNull(esSearchRequest.getHighlightBuilder())) {
                searchSourceBuilder.highlighter(esSearchRequest.getHighlightBuilder());
            }
            //排序字段
            if (StringUtils.isNotBlank(sortName)) {
                searchSourceBuilder.sort(sortName);
            }
            //排序类型为降序
            searchSourceBuilder.sort(new ScoreSortBuilder().order(SortOrder.DESC));

            // 创建一个SearchRequest对象，用于构建搜索请求
            SearchRequest searchRequest = new SearchRequest();
            // 设置搜索类型为默认类型
            searchRequest.searchType(SearchType.DEFAULT);
            // 指定要搜索的索引名称
            searchRequest.indices(esIndexInfo.getIndexName());
            // 设置搜索请求的源，即之前构建的SearchSourceBuilder对象，包含了查询条件、排序、高亮等设置
            searchRequest.source(searchSourceBuilder);
            // 如果需要快照
            if (needScroll) {
                // 创建一个Scroll对象，指定滚动的时间间隔
                Scroll scroll = new Scroll(TimeValue.timeValueMinutes(minutes));
                // 将滚动参数设置到搜索请求中
                searchRequest.scroll(scroll);
            }
            // 使用之前获取的RestHighLevelClient实例执行搜索请求，返回一个SearchResponse对象，包含了搜索结果
            SearchResponse search = getClient(esIndexInfo.getClusterName()).search(searchRequest, COMMON_OPTIONS);
            return search;
        } catch (Exception e) {
            log.error("searchWithTermQuery.exception:{}", e.getMessage(), e);
        }
        return null;
    }

    public static boolean batchInsertDoc(EsIndexInfo esIndexInfo, List<EsSourceData> esSourceDataList) {
        if (log.isInfoEnabled()) {
            log.info("批量新增ES:" + esSourceDataList.size());
            log.info("indexName:" + esIndexInfo.getIndexName());
        }
        try {
            boolean flag = false;
            BulkRequest bulkRequest = new BulkRequest();

            for (EsSourceData source : esSourceDataList) {
                String docId = source.getDocId();
                if (StringUtils.isNotBlank(docId)) {
                    IndexRequest indexRequest = new IndexRequest(esIndexInfo.getIndexName());
                    indexRequest.id(docId);
                    indexRequest.source(source.getData());
                    bulkRequest.add(indexRequest);
                    flag = true;
                }
            }


            if (flag) {
                BulkResponse response = getClient(esIndexInfo.getClusterName()).bulk(bulkRequest, COMMON_OPTIONS);
                if (response.hasFailures()) {
                    return false;
                }
            }
        } catch (Exception e) {
            log.error("batchInsertDoc.error", e);
        }

        return true;
    }

    public static boolean updateByQuery(EsIndexInfo esIndexInfo, QueryBuilder queryBuilder, Script script, int batchSize) {
        if (log.isInfoEnabled()) {
            log.info("updateByQuery.indexName:" + esIndexInfo.getIndexName());
        }
        try {
            UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest(esIndexInfo.getIndexName());
            updateByQueryRequest.setQuery(queryBuilder);
            updateByQueryRequest.setScript(script);
            updateByQueryRequest.setBatchSize(batchSize);
            updateByQueryRequest.setAbortOnVersionConflict(false);
            BulkByScrollResponse response = getClient(esIndexInfo.getClusterName()).updateByQuery(updateByQueryRequest, RequestOptions.DEFAULT);
            List<BulkItemResponse.Failure> failures = response.getBulkFailures();
        } catch (Exception e) {
            log.error("updateByQuery.error", e);
        }
        return true;
    }

    /**
     * 分词方法
     */
    public static List<String> getAnalyze(EsIndexInfo esIndexInfo, String text) throws Exception {
        List<String> list = new ArrayList<String>();
        Request request = new Request("GET", "_analyze");
        JSONObject entity = new JSONObject();
        entity.put("analyzer", "ik_smart");
        entity.put("text", text);
        request.setJsonEntity(entity.toJSONString());
        Response response = getClient(esIndexInfo.getClusterName()).getLowLevelClient().performRequest(request);
        JSONObject tokens = JSONObject.parseObject(EntityUtils.toString(response.getEntity()));
        JSONArray arrays = tokens.getJSONArray("tokens");
        for (int i = 0; i < arrays.size(); i++) {
            JSONObject obj = JSON.parseObject(arrays.getString(i));
            list.add(obj.getString("token"));
        }
        return list;
    }


}
